package com.bj58.ecdata.calc.dbservice.handler;

import java.util.Date;
import java.util.concurrent.ExecutorService;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import com.bj58.ecdata.calc.dbservice.consts.Constant;
import com.bj58.ecdata.calc.dbservice.dao.model.ClickModel;
import com.bj58.ecdata.calc.dbservice.dao.model.DisplayModel;
import com.bj58.ecdata.calc.dbservice.dao.model.SearchWord;
import com.bj58.ecdata.calc.dbservice.dao.model.TrackModel_TMP;
import com.bj58.ecdata.calc.dbservice.service.DataOptService;
import com.bj58.ecdata.redis.JedisX;
import com.bj58.ecdata.thread.ThreadUtils;
import com.bj58.ecdata.util.DateUtils;
import com.bj58.ecdata.util.JsonUtils;

/**
 * 主服务类
 * @author zhaoxiang
 */
@Component
public class DataInsertHandler implements InitializingBean {

	private Logger log = LoggerFactory.getLogger(DataInsertHandler.class);

	@Autowired(required = false)
	@Qualifier("cachePool_handler")
	protected JedisX cache;  //spark输出redis

	@Autowired(required = false)
	@Qualifier("cachePool")
	protected JedisX cache_result; //结果缓存redis
	
	@Autowired(required = false)
	@Qualifier("cookie_pool_cache")
	protected JedisX cookie_cache; //cookie结果缓存redis
	
	@Autowired
	private DataOptService dataOptService;

	class CpcClickTask implements Runnable {

		@Override
		public void run() {
			while (true) {
				try {
					String str = cache.rpopString(Constant.CPC_CLICK_REDIS_QUEUE_KEY);
					if (StringUtils.isNotBlank(str)) {
						CpcClickModel model = JsonUtils.fromJSON(str, CpcClickModel.class);
						if (null != model) {
							ClickModel cm = new ClickModel();
							cm.setBussline(model.getBusiLine());
							cm.setCash(model.getCash());
							cm.setCate2Id(model.getCate2Id());
							cm.setCate2Name(model.getCate2Name());
							cm.setCity1Id(model.getCity1Id());
							cm.setCity1Name(model.getCity1Name());
							cm.setCityType(model.getCityType());
							cm.setCost(model.getCost());
							cm.setCustNum(model.getCustNum());
							cm.setPlatform(model.getPlatform());
							cm.setPv(model.getPv());
							cm.setTime(model.getTime());
							dataOptService.insertClick(cm);
						}
					} else
						Thread.sleep(100);
				} catch (Exception e) {
					log.error("insert click error!", e);
				}
			}

		}

	}

	class CpcDisplayTask implements Runnable {

		@Override
		public void run() {
			while (true) {
				try {
					String str = cache.rpopString(Constant.CPC_DISPLAY_REDIS_QUEUE_KEY);
					if (StringUtils.isNotBlank(str)) {
						CpcDisplayModel model = JsonUtils.fromJSON(str, CpcDisplayModel.class);
						if (null != model) {
							DisplayModel dm = new DisplayModel();
							dm.setAd(model.getAdPv());
							dm.setAdNum(model.getAdNum());
							dm.setBussline(model.getBusiLine());
							dm.setPlatform(model.getPlatform());
							dm.setPv(model.getPv());
							dm.setTime(model.getTime());
							dataOptService.insertDisplay(dm);
						}
					} else
						Thread.sleep(100);
				} catch (Exception e) {
					log.error("insert display error!", e);
				}
			}

		}

	}
	
	class TrackSearchWordTask implements Runnable{

		@Override
		public void run() {
			while (true) {
				try {
					String str = cache.rpopString(Constant.SEARCH_WORD_TRACK);
					if (StringUtils.isNotBlank(str)) {
						TrackSearchWord model = JsonUtils.fromJSON(str, TrackSearchWord.class);
						if (null != model) {
							SearchWord sw = new SearchWord();
							sw.setBussline(model.getBusiLine());
							sw.setCount(1l);
							sw.setPlatform(model.getPlatform());
							sw.setSearchWord(model.getSearchWord());
							sw.setTime(DateUtils.getTruncatedToday());
							dataOptService.insertSearchWord(sw);
						}
					} else
						Thread.sleep(100);
				} catch (Exception e) {
					log.error("insert track search word error!", e);
				}
			}
		}
		
	}

	class TrackTask implements Runnable {

		@Override
		public void run() {
			while (true) {
				try {
					String str = cache.rpopString(Constant.TRACK_REDIS_QUEUE_KEY);
					if (StringUtils.isNotBlank(str)) {
						TrackModel model = JsonUtils.fromJSON(str, TrackModel.class);
						if (null != model) {
							com.bj58.ecdata.calc.dbservice.dao.model.TrackModel tm = new com.bj58.ecdata.calc.dbservice.dao.model.TrackModel();
							tm.setBussline(model.getBusiLine());
							tm.setListpv(model.getListpv());
							tm.setPlatform(model.getPlatform());
							tm.setPv(model.getPv());
							tm.setTime(model.getTime());
							tm.setUv(model.getUv());
							tm.setVppv(model.getVppv());
							dataOptService.insertTrack(tm);
						}
					} else
						Thread.sleep(100);
				} catch (Exception e) {
					log.error("insert track error!", e);
				}
			}
		}

	}
	
	class CookieTask implements Runnable {
		@Override
		public void run() {
			final long interval = 10 * 1000;//每10秒 更新一次UV
			String[] COOKIE_REDIS_KEYS = {Constant.COOKIE_REDIS_QUEUE_KEY_PC, Constant.COOKIE_REDIS_QUEUE_KEY_M, Constant.COOKIE_REDIS_QUEUE_KEY_APP};
			int[] expiredTime = {Constant.redis_1h, Constant.redis_2h, Constant.redis_1h*3};
			String[] UV_REDIS_KEYS = {"uv_-1_%s", "uv_0_%s", "uv_2_%s"};

			while (true) {
				try {
					long time = new Date().getTime();
					long curMinTime = DateUtils.getTruncatedMinute();
					String date = DateFormatUtils.format(curMinTime, "yyyyMMdd");
					long allUV = 0;
					for(int i=0; i < COOKIE_REDIS_KEYS.length; i++){
						String cookie_redis_key = String.format(COOKIE_REDIS_KEYS[i], date);
						String uv_redis_key = String.format(UV_REDIS_KEYS[i], curMinTime);
						//取cookie集大小
						long curUV = 0;
						curUV = cookie_cache.scard(cookie_redis_key);
						allUV += curUV;
						cache_result.setString(uv_redis_key, Constant.redis_2d, Long.toString(curUV));
//						cache.expire(cookie_redis_key, expiredTime[i]); //当天cookie第二天 2小时后删除
						cookie_cache.expire(cookie_redis_key, expiredTime[i]); //当天cookie第二天 2小时后删除
					}
					cache_result.setString(String.format("uv_-2_%s", curMinTime), Constant.redis_2d, Long.toString(allUV));
					cache_result.setString("uv_latest_time",0, Long.toString(time));
					Thread.sleep(interval);
				} catch (Exception e) {
					log.error("insert UV error!", e);
				}
			}
		}
		
	}

	class TrackTask_TMP implements Runnable {

		@Override
		public void run() {
			while (true) {
				try {
					String str = cache.rpopString(Constant.TRACK_REDIS_QUEUE_KEY_tmp);
					if (StringUtils.isNotBlank(str)) {
						TrackModel_TMP model = JsonUtils.fromJSON(str, TrackModel_TMP.class);
						if (null != model) {
							dataOptService.insertTrack_TMP(model);
						}
					} else
						Thread.sleep(100);
				} catch (Exception e) {
					log.error("insert track error!", e);
				}
			}
		}

	}

	class InfoTask implements Runnable {

		@Override
		public void run() {
			while (true) {
				try {
					String str = cache.rpopString(Constant.INFO_REDIS_QUEUE_KEY);
					if (StringUtils.isNotBlank(str)) {
						InfoModel model = JsonUtils.fromJSON(str, InfoModel.class);
						if (null != model) {
							com.bj58.ecdata.calc.dbservice.dao.model.InfoModel im = new com.bj58.ecdata.calc.dbservice.dao.model.InfoModel();
							im.setTime(model.getTime());
							im.setPlatform(model.getPlatform());
							im.setBussline(model.getBusiLine());
							im.setBelong_cate1(model.getBelong_cate1());
							im.setBelong_cate2(model.getBelong_cate2());
							im.setInfoNum(model.getInfoNum());
							dataOptService.insertInfo(im);
						}
					} else
						Thread.sleep(200);
				} catch (Exception e) {
					log.error("insert info error!", e);
				}
			}
		}

	}
	
	class ResumeAddTask implements Runnable {
		
		@Override
		public void run() {
			while (true) {
				try {
					String str = cache.rpopString(Constant.RESUME_ADD_REDIS_QUEUE_KEY);
					if (StringUtils.isNotBlank(str)) {
							ResumeAddModel model = JsonUtils.fromJSON(str, ResumeAddModel.class);
						if (null != model) {
							com.bj58.ecdata.calc.dbservice.dao.model.ResumeAddModel im = new com.bj58.ecdata.calc.dbservice.dao.model.ResumeAddModel();
							im.setTime(model.getTime());
							im.setPlatform(model.getPlatform());
							im.setCity1Id(model.getTargetAreaID());
							im.setCity1Name(model.getTargetAreaName());
							im.setCateId(model.getTargetCateID());
							im.setCate1Name(model.getTargetCateName1());
							im.setCate2Name(model.getTargetCateName2());
							im.setAddNum(model.getNum());
							dataOptService.insertResumeAdd(im);
						}
					} else
						Thread.sleep(300);
				} catch (Exception e) {
					log.error("insert info error!", e);
				}
			}
		}
		
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		ExecutorService executorService = ThreadUtils.newFixedThreadPool(7, "queue_threads");
		executorService.submit(new TrackTask());
		executorService.submit(new CpcDisplayTask());
		executorService.submit(new CpcClickTask());
		executorService.submit(new TrackSearchWordTask());
		executorService.submit(new InfoTask());
		executorService.submit(new ResumeAddTask());
		executorService.submit(new CookieTask());
	}

	/**
	 * @return the dataOptService
	 */
	public DataOptService getDataOptService() {
		return dataOptService;
	}

	/**
	 * @param dataOptService
	 *            the dataOptService to set
	 */
	public void setDataOptService(DataOptService dataOptService) {
		this.dataOptService = dataOptService;
	}

}
